package org.apm.data.client;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apm.data.util.TimeUtil;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class ClientFactory {
    private static final Logger LOGGER = LogManager.getLogger(ClientFactory.class);

    private static final Map<String, Client> CLIENTS = new ConcurrentHashMap<>();
    private static final Map<String, BulkProcessor> BULK_PROCESSOR_MAP = new ConcurrentHashMap<>();

    // 可调控参数
    private static final ByteSizeValue bulkSize = new ByteSizeValue(10, ByteSizeUnit.MB);
    private static final TimeValue bulkTime = TimeValue.timeValueSeconds(3);
    private static final int concurrentRequest = 64;
    private static final int bulkQuantity = 20000;

    public static Client getClient(final String clusterName, final Map<String, Integer> hosts) {
        String id = clusterName + "#" + hosts;
        synchronized (CLIENTS) {
            if (null == CLIENTS.get(id)) {
                Client client = initClient(clusterName, hosts);
                CLIENTS.put(id, client);
                BULK_PROCESSOR_MAP.put(id, initBulkProcessor(client));
            }
        }
        return CLIENTS.get(id);
    }

    public static BulkProcessor getBulkProcessor(final String clusterName, final Map<String, Integer> hosts) {
        String id = clusterName + "#" + hosts;
        synchronized (CLIENTS) {
            getClient(clusterName,hosts);
        }
        return BULK_PROCESSOR_MAP.get(id);
    }

    private static Client initClient(final String clusterName, final Map<String, Integer> hosts) {
        Settings settings = Settings.builder().put("cluster.name", clusterName).build();
        TransportClient transportClient = new PreBuiltTransportClient(settings);
        for (String ip : hosts.keySet()) {
            try {
                transportClient
                        .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ip), hosts.get(ip)));
            } catch (UnknownHostException exception) {
                LOGGER.error("UnknownHostException, " + ip + " can't be transformed to host.", exception);
            }
        }
        return transportClient;
    }

    private static BulkProcessor initBulkProcessor(final Client client) {
        BulkProcessor esBulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() {
            private int number = 0;
            private int context = 0;
            private long start = 0;

            @Override
            public void beforeBulk(final long executionId, final BulkRequest request) {
                start = System.currentTimeMillis();
                context = request.requests().size();
                number = request.numberOfActions();
            }

            @Override
            public void afterBulk(final long executionId, final BulkRequest request, final BulkResponse response) {
                long end = System.currentTimeMillis();
                StringBuilder stringBuilder = new StringBuilder("[Bulk][").append(executionId).append("] 始于: ")
                        .append(TimeUtil.timestampToDate(start)).append(", 终于: ").append(TimeUtil.timestampToDate(end))
                        .append(", took: ").append(response.getTookInMillis()).append(", ").append(number).append(", ")
                        .append(context).append(", ").append(response.buildFailureMessage());
                LOGGER.info(stringBuilder.toString());
            }

            @Override
            public void afterBulk(final long executionId, final BulkRequest request, final Throwable failure) {
                StringBuilder stringBuilder = new StringBuilder("[Bulk][").append(executionId).append("] ")
                        .append(failure.getMessage());
                LOGGER.error(stringBuilder);
            }
        }).setBulkActions(bulkQuantity).setBulkSize(bulkSize).setFlushInterval(bulkTime)
                .setConcurrentRequests(concurrentRequest).build();

        return esBulkProcessor;
    }
}
